/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.subThread;

import cn.ac.iie.di.tools.FileTools;
import cn.ac.iie.di.tools.FtpTools;
import cn.ac.iie.di.tools.PropsFiles;
import cn.ac.iie.di.tools.StringTools;
import cn.ac.iie.di.tools.XmlTools;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.log4j.Logger;

/**
 *
 * @author root
 */
public class AgentThread implements Runnable {

    private static Logger logger = Logger.getLogger(AgentThread.class);
    public static String requestPath = ClassLoader.getSystemClassLoader().getResource("requestXML.xml").getFile();
    public String xml;
    public String name;
    public String hdfsPath;//xml中的根目录
    public List<String> agentIPList = new ArrayList<String>();//ip:port
    public String subjobid;
    public long timeout;
    public String hdfsPathForRDD;//拼凑多级目录
    public String responseXml;

    public AgentThread(String xml, String subjobid) {
        this.xml = xml;
        this.subjobid = subjobid;
    }

    @Override
    public void run() {
        //==============================初始化参数，并进行验证==================================
        String responseXml = FileTools.readFile(requestPath);
        try {
            parse(xml);
        } catch (Exception ex) {
            String code = "1";
            String message = ex.getMessage();
            logger.info(subjobid + " " + message);
            responseXml = StringTools.replaceAll(responseXml, subjobid, code, message);
            logger.info(subjobid + " responseXml:" + responseXml);
//            new JettyClient().send(PropsFiles.getValue("schedulerPath"), responseXml, subjobid);
            return;
        }

        //============================启动各个实际代理===========================================
        long allAgentBeginTime = System.currentTimeMillis();
        logger.info(subjobid + " dataInputToRDD begin!");
        try {
            logger.info(subjobid + " dataInputToRDD runAgent begin!");

            //开始运行每一个代理，并且等待代理返回信息，只要有一个代理部分成功，就算成功
            runAgent();
        } catch (Exception ex) {
            logger.info(subjobid + " " + ex.getMessage());
            String code = "1";
            String message = ex.getMessage();
            logger.info(subjobid + " " + message);
            responseXml = StringTools.replaceAll(responseXml, subjobid, code, message);
            logger.info(subjobid + " responseXml:" + responseXml);
//            new JettyClient().send(PropsFiles.getValue("schedulerPath"), responseXml, subjobid);
            return;
        }

        long allAgentEndTime = System.currentTimeMillis();
        long rumtime = (allAgentEndTime - allAgentBeginTime) / 1000;
        logger.info(subjobid + " all Agent runTime is " + rumtime);
        logger.info(subjobid + " dataInputToRDD runAgent end!");
        //================================将文件路径写入hive表=================================
        String tableName = "";
        tableName = XmlTools.getValueFromStrDGText(xml, "outport1");
        boolean flag = insertTmpTable(tableName);
        if (!flag) {
            logger.info(subjobid + " insert to hive error!");
            String code = "1";
            String message = subjobid + " insert to hive error!";
            logger.info(subjobid + " " + message);
            responseXml = StringTools.replaceAll(responseXml, subjobid, code, message);
            logger.info(subjobid + " responseXml:" + responseXml);
//            new JettyClient().send(PropsFiles.getValue("schedulerPath"), responseXml, subjobid);
            return;
        }

        //================================返回正确xml========================================
        String code = "0";
        String message = "nothing";
        responseXml = StringTools.replaceAll(responseXml, subjobid, code, message);
        logger.info(subjobid + " responseXml:" + responseXml);
//        new JettyClient().send(PropsFiles.getValue("schedulerPath"), responseXml, subjobid);
        logger.info(subjobid + " sucess send: " + responseXml);
        return;
    }

    public void parse(String xml) throws Exception {
        //解析xml，并验证
        long parseBeginTime = System.currentTimeMillis();
        //得到任务号
        subjobid = XmlTools.getElValueFromStr(xml, "subjobid");
        if (subjobid.equals("")) {
            throw new Exception("subjobid is empty!");
        }
        logger.info(subjobid + " parse begin!");
        //得到超时时间
        String timeout = XmlTools.getValueFromStrDGText(xml, "timeout");
        if (timeout.equals("")) {
            logger.info(subjobid + " timeout is empty!");
            throw new Exception(subjobid + " timeout is empty!");
        }
        try {
            this.timeout = Long.parseLong(timeout);
        } catch (Exception ex) {
            logger.info(subjobid + " timeout is not a long!");
            throw new Exception(subjobid + " timeout is not a long!");
        }
        //得到系统名称
        name = XmlTools.getValueFromStrDGText(xml, "name");
        logger.info(subjobid + " name is " + name);
        if (name.equals("")) {
            logger.info(subjobid + " name is empty!");
            throw new Exception(subjobid + " name is empty!");
        }
        //得到上传文件的路径
        hdfsPath = XmlTools.getElement3AttributeValue(xml, "operator", "context", "property", "tempHDFSPath");
        logger.info(subjobid + " hdfsPath is " + hdfsPath);
        if (hdfsPath.equals("")) {
            logger.info(subjobid + " hdfsPath is empty!");
            throw new Exception(subjobid + " hdfsPath is empty!");
        }
        //得到待发送的代理ip
        String agentIPstr = XmlTools.getValueFromStrDGText(xml, "agentIP");
        if (agentIPstr.equals("")) {
            logger.info(subjobid + " agentIP is empty!");
            throw new Exception(subjobid + " agentIP is empty!");
        }
        logger.info(subjobid + " agentIPstr is " + agentIPstr);
        String ls[] = agentIPstr.split("[|]");
        for (int i = 0; i < ls.length; i++) {
            agentIPList.add(ls[i]);
        }
        //赋值
        this.xml = xml;
        Date now = new Date();
        SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd");//可以方便地修改日期格式
        String today = dateFormat.format(now);
        this.hdfsPathForRDD = hdfsPath + "/" + name + "/" + today + "/" + subjobid;
        logger.info(subjobid + " hdfsPathForRDD is " + hdfsPathForRDD);
        long parseEndTime = System.currentTimeMillis();
        long runtime = (parseEndTime - parseBeginTime) / 1000;
        logger.info(subjobid + " parse xml runtime is " + runtime + " s");
        logger.info(subjobid + " parse end!");
    }

    public void runAgent() throws Exception {

        String type = XmlTools.getValueFromStrDGText(xml, "type");

        BlockingQueue<String> rs_xml = new LinkedBlockingQueue<String>();//各个代理状态
        //一个代理一个线程，开始实际采集
        ThreadPoolExecutor threadPool = new ThreadPoolExecutor(Integer.parseInt(PropsFiles.getValue("coreThreadPoolSize")), Integer.parseInt(PropsFiles.getValue("maxThreadPoolSize")), 3,
                TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(Integer.parseInt(PropsFiles.getValue("workQueueSize"))),
                new ThreadPoolExecutor.DiscardOldestPolicy());

        if ("ftpupload".equals(type)) {
            for (int i = 0; i < agentIPList.size(); i++) {
                threadPool.execute(new AgentUpload(agentIPList.get(i), xml, subjobid, hdfsPathForRDD, rs_xml));
            }
        } else if ("ftpdownload".equals(type)) {
            //ftp信息
            String ip = XmlTools.getValueFromStrDGText(xml, "ip");
            String port = XmlTools.getValueFromStrDGText(xml, "port");
            String username = XmlTools.getValueFromStrDGText(xml, "username");
            String password = XmlTools.getValueFromStrDGText(xml, "password");
            String filepath = XmlTools.getValueFromStrDGText(xml, "filepath");
            logger.info("the receive ftp's message is[ip: " + ip + ",port: " + port + ",username: " + username + ",password: " + password + ",filepath: " + filepath + "]");

            String[] str = filepath.split("/");
            String namespace = str[str.length - 1];//0_shipin
            String realnamespace = namespace.substring(namespace.indexOf("_") + 1, namespace.length());//shipin
            logger.info("the namespace is: " + namespace + " ,and the realnamespace is: " + realnamespace);

            //读取Ctrl目录下指定量的json
            List<String> jsonList = new FtpTools().getJsonFiles_new(ip, Integer.parseInt(port), username, password, filepath + "/Ctrl", "", realnamespace);
            // List转换为Set,去除json中重复的文件名
            HashSet<String> jsonSet = new HashSet<String>(jsonList);
            logger.info("============================the jsonSet size " + jsonSet.size());
            jsonList.clear();
            jsonList.addAll(jsonSet);

            //把jsonList平均分配给agentList
            List<String> datalist = new ArrayList<String>();

            if (!jsonList.isEmpty()) {
                if (jsonList.size() < agentIPList.size()) {
                    for (int h = 0; h < jsonList.size(); h++) {
                        String jsonString = jsonList.get(h);
                        datalist.add(jsonString);
                    }
                } else {
                    int count = jsonList.size() / agentIPList.size();
                    int n = 0;
                    for (int m = 1; m <= agentIPList.size(); m++) {
                        String jsonString = "";
                        for (int j = n; j < count * m; j++) {
                            jsonString += jsonList.get(j) + "|";
                            n++;
                        }

                        if (m == agentIPList.size()) {
                            for (int k = n; k < jsonList.size(); k++) {
                                jsonString += jsonList.get(k) + "|";
                            }
                            datalist.add(jsonString);
                        } else {
                            datalist.add(jsonString);
                        }
                    }
                }
            } else {
                logger.info("======================ftp上没有可以执行的JSON文件========================");
            }

            for (int i = 0; i < datalist.size(); i++) {
                threadPool.execute(new Agent(agentIPList.get(i), xml, subjobid, hdfsPathForRDD, rs_xml, datalist.get(i)));
            }
        }

        //全部代理采集完成了，任务结束，销毁线程池，超时也销毁
        Thread.sleep(3000);
        long begin = System.currentTimeMillis();
        while (true) {
            int activeThread = threadPool.getActiveCount();
            if (activeThread == 0) {
                break;
            } else {
                long end = System.currentTimeMillis();
                long middle = (end - begin) / 1000;
                if (middle < timeout) {
                    logger.info(subjobid + " runtime:" + middle + " timeout:" + timeout);
                    Thread.sleep(3000);
                } else {
                    logger.info(subjobid + " job is killed, because of timeout!");
                    throw new Exception(subjobid + " job is killed, because of timeout!");
                }

            }
        }
        threadPool.shutdown();

        //解析各个代理返回结果状态
        //没开启的代理不返回东西，因此，通过返回结果的个数与代理ip的个数可以判断出哪个代理没返回
        if (rs_xml.size() < agentIPList.size()) {
            logger.info(subjobid + " some agnets have no return!send error!");
            List<String> errAgentIP = new ArrayList<String>();
            for (int j = 0; j < agentIPList.size(); j++) {
                boolean flag = false;
                Iterator<String> itr1 = rs_xml.iterator();
                while (itr1.hasNext()) {
                    String agentXml1 = itr1.next();
                    String agentIP = XmlTools.getElValueFromStr(agentXml1, "agentIP");
                    if (agentIP.equals(agentIPList.get(j))) {
                        flag = true;
                        break;
                    }
                }
                if (!flag) {
                    errAgentIP.add(agentIPList.get(j));
                }

            }

            String errAgentIPstr = "";
            for (int k = 0; k < errAgentIP.size(); k++) {
                errAgentIPstr = errAgentIPstr + "|" + errAgentIP.get(k);
            }
            logger.info(subjobid + "###" + errAgentIPstr + " have no return!send error!");
            throw new Exception(subjobid + "###" + errAgentIPstr + " have no return!send error!");
        }
        Iterator<String> itr = rs_xml.iterator();
        logger.info(subjobid + "return begin ================================================");
        //boolean sucessFlag = true;//本次任务是否成功的标志
        String totalXml = "";//要是不成功，就把这个xml报出去。这是全部xml拼接成的。
        String errXML = "";
        while (itr.hasNext()) {
            String agentXml = itr.next();
            totalXml = totalXml + "\r\n" + agentXml;
            logger.info(subjobid + ":" + agentXml);
            String instance = XmlTools.getElValueFromStr(agentXml, "instance");
            String operatorname = XmlTools.getElValueFromStr(agentXml, "operatorname");
            String status = XmlTools.getElValueFromStr(agentXml, "status");
            String desc = XmlTools.getElValueFromStr(agentXml, "desc");
            String agentIP = XmlTools.getElValueFromStr(agentXml, "agentIP");
            if (status.equals("fail") || status.equals("half_sucess")) {
                logger.info(subjobid + " agentIP:" + agentIP + " status:" + status + ":" + desc);
                errXML = errXML + operatorname + "|" + instance + "|" + desc + "|" + agentIP + ":" + status + ",";
//                sucessFlag = false;
//                break;
            }
            if (!errXML.isEmpty()) {
                errXML = errXML.substring(0, errXML.length() - 1);
            }
        }
        logger.info(subjobid + "return end ================================================");
        if (errXML.contains("fail") || errXML.contains("half_sucess")) {
            logger.info(subjobid + " some agent are error! totalXml" + totalXml);
            logger.info(subjobid + " some agent are error! " + errXML);
            throw new Exception(errXML);
        } else {
            logger.info(subjobid + " all agent are sucess! " + totalXml);
        }
    }

    public boolean insertTmpTable(String tableName) {
        return true;
    }
}
